package com.itcast.spark.basePro

import org.apache.spark.mllib.linalg
import org.apache.spark.mllib.linalg.{Matrix, Vectors}
import org.apache.spark.mllib.stat.{MultivariateStatisticalSummary, Statistics}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.immutable

/**
 * DESC:
 */
object _02IrisSelectFeaturesVarience {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("_01IrisVarience").setMaster("local[*]")
    val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.setLogLevel("WARN")
    //读取数据
    val dataRDD: RDD[linalg.Vector] = sc.textFile("./datasets/mldata/irisNoHeader.csv")
      .map(_.split(",")(0))
      .map(_.toDouble)
      .map(x => Vectors.dense(x))
    //使用Summary
    val summary: MultivariateStatisticalSummary = Statistics.colStats(dataRDD)
    println("count is:", summary.count)
    println("nonzoreos is:", summary.numNonzeros)
    println("max is:", summary.max)
    println("min is:", summary.min)
    println("mean is:", summary.mean)
    println("varience is:", summary.variance)
    /*  (count is:,150)
      (nonzoreos is:,[150.0])
      (max is:,[7.9])
      (min is:,[4.3])
      (mean is:,[5.843333333333333])
      (varience is:,[0.6856935123042512])*/

    val rdd1: RDD[Double] = sc.textFile("./datasets/mldata/irisNoHeader.csv")
      .map(_.split(",")(0))
      .map(_.toDouble)
    val rdd2: RDD[Double] = sc.textFile("./datasets/mldata/irisNoHeader.csv")
      .map(_.split(",")(1))
      .map(_.toDouble)

    val corr: Double = Statistics.corr(rdd1, rdd2)
    println("corr value is:", corr)

    //读取数据---所有的特征的数据
    val dataRDD1: RDD[linalg.Vector] = sc.textFile("./datasets/mldata/irisNoHeader.csv")
      .map(_.split(","))
      .map(x => (x(0).toDouble, x(1).toDouble, x(2).toDouble, x(3).toDouble))
      .map(x => Vectors.dense(x._1, x._2, x._3, x._4))
    val corrMatrix: Matrix = Statistics.corr(dataRDD1)
    println("corr value is:", corrMatrix)

    val rdd3 = sc.textFile("./datasets/mldata/irisNoHeader.csv")
      .map(line => (line.split(",")(0),line.split(",")(3)))
  }
}
